package utils

import (
	"context"
	"database/sql"
	"fmt"
	"github.com/go-sql-driver/mysql"
	"math"
	"runtime"
	"strings"
	"sync"
	"time"
)

// MysqlDBs 存放连接
type MysqlDBs struct {
	kv map[string]*sql.DB
	l  sync.RWMutex
}

// MysqlRow 数据单元
type MysqlRow map[string]interface{}

// QueryLog 查询日志结构体
type QueryLog struct {
	When   string
	Query  string
	Params []interface{}
	Err    error
}

// MysqlConnOption 单库连接配置
type MysqlConnOption struct {
	Dsn             string
	Addr            string
	DBName          string
	MaxOpenConn     int
	MaxIdleConn     int
	ConnMaxIdleTime time.Duration
	ConnMaxLifetime time.Duration
}

// MysqlConnParam 连接配置
type MysqlConnParam struct {
	Master *MysqlConnOption
	Slaves []*MysqlConnOption
}

// MysqlLog 操作日志
type MysqlLog struct {
	When   string
	Query  string
	Params []interface{}
	Err    error
}

// Mysql 连接实例
type Mysql struct {
	ConnParam     *MysqlConnParam
	MasterAsSlave bool
	TimeoutInsert time.Duration
	TimeoutSelect time.Duration
	TimeoutUpdate time.Duration
	TimeoutTrans  time.Duration
	Logger        *Queue
	Listener      func(err error, query string, param ...interface{})
}

var mysqlDBs *MysqlDBs // 保存连接的变量

// 初始化
func init() {
	mysqlDBs = &MysqlDBs{
		kv: make(map[string]*sql.DB),
	}
}

// NewMysqlConnOption 得到一个单库连接配置
func NewMysqlConnOption(dsn string, maxOpen, maxIdle int, maxIdleTime, maxLifeTime time.Duration) (*MysqlConnOption, error) {
	addr := ""
	dbName := ""
	if cfg, err := mysql.ParseDSN(dsn); err != nil {
		return nil, err
	} else if cfg.Addr == "" || cfg.DBName == "" {
		return nil, fmt.Errorf("empty Addr or DBName")
	} else {
		addr = cfg.Addr
		dbName = cfg.DBName
	}
	if maxOpen <= 0 {
		maxOpen = runtime.NumCPU()*2 + 1
	}
	if maxIdle <= 0 {
		maxIdle = maxOpen / 2
		if maxIdle < 2 {
			maxIdle = 2
		}
	}
	if maxIdleTime < 0 {
		maxIdleTime = 3 * time.Minute
	}
	if maxLifeTime < 0 {
		maxLifeTime = 30 * time.Minute
	}
	return &MysqlConnOption{
		Dsn:             dsn,
		Addr:            addr,
		DBName:          dbName,
		MaxOpenConn:     maxOpen,
		MaxIdleConn:     maxIdle,
		ConnMaxIdleTime: maxIdleTime,
		ConnMaxLifetime: maxLifeTime,
	}, nil
}

// NewMysql 得到一个连接实例
func NewMysql(param *MysqlConnParam) *Mysql {
	logger, _ := NewQueue(20)
	return &Mysql{
		ConnParam:     param,
		MasterAsSlave: len(param.Slaves) == 0,
		TimeoutSelect: 3 * time.Second,
		TimeoutInsert: 3 * time.Second,
		TimeoutUpdate: 3 * time.Second,
		TimeoutTrans:  5 * time.Second,
		Logger:        logger,
	}
}

// SetListener 设置监听函数
func (m *Mysql) SetListener(fn func(err error, query string, param ...interface{})) {
	m.Listener = fn
}

// CloseDB 关闭连接
func (m *Mysql) CloseDB() {
	mysqlDBs.l.Lock()
	defer mysqlDBs.l.Unlock()
	connOptions := append(m.ConnParam.Slaves, m.ConnParam.Master)
	for _, co := range connOptions {
		k := co.Addr + ":" + co.DBName
		if db, ok := mysqlDBs.kv[k]; ok {
			if db != nil {
				_ = db.Close()
			}
			delete(mysqlDBs.kv, k)
		}
	}
}

// GetDB 获取一个连接
func (m *Mysql) GetDB(useSlave bool) (*sql.DB, error) {
	if useSlave && m.MasterAsSlave {
		useSlave = false
	}
	var p *MysqlConnOption
	var err error
	if !useSlave {
		p = m.ConnParam.Master
	} else if l := len(m.ConnParam.Slaves); l > 0 {
		p = m.ConnParam.Slaves[RandInt(l)]
	} else {
		p = m.ConnParam.Slaves[0]
	}
	if p == nil || p.Dsn == "" || p.Addr == "" || p.DBName == "" {
		return nil, fmt.Errorf("there is a wrong connection option")
	}
	k := p.Addr + ":" + p.DBName
	mysqlDBs.l.RLock()
	db, ok := mysqlDBs.kv[k]
	mysqlDBs.l.RUnlock()
	if ok && db != nil {
		return db, nil
	}
	mysqlDBs.l.Lock()
	defer mysqlDBs.l.Unlock()
	if db, ok = mysqlDBs.kv[k]; ok && db != nil {
		return db, nil
	}
	if db, err = sql.Open("mysql", p.Dsn); err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), m.TimeoutSelect)
	defer cancel()
	if err = db.PingContext(ctx); err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(p.MaxOpenConn)
	db.SetMaxIdleConns(p.MaxIdleConn)
	db.SetConnMaxIdleTime(p.ConnMaxIdleTime)
	db.SetConnMaxLifetime(p.ConnMaxLifetime)
	if !ok {
		mysqlDBs.kv[k] = db
	}
	return db, nil
}

// Truncate 清空一张表
func (m *Mysql) Truncate(table string) error {
	db, err := m.GetDB(false)
	if err != nil {
		return err
	}
	q := "TRUNCATE TABLE `" + table + "`"
	ctx, cancel := context.WithTimeout(context.Background(), m.TimeoutUpdate)
	defer cancel()
	defer func() {
		m.AddLog(q, nil, err)
	}()
	_, err = db.ExecContext(ctx, q)
	return err
}

// InsertDuplicate 插入，如果存在就更新
func (m *Mysql) InsertDuplicate(table string, row *MysqlRow) (int64, error) {
	if row == nil || *row == nil || len(*row) == 0 {
		return 0, fmt.Errorf("row is empty")
	}
	cols := make([]string, 0)
	phds := make([]string, 0)
	sets := make([]string, 0)
	params := make([]interface{}, 0)
	for col, param := range *row {
		cols = append(cols, col)
		params = append(params, param)
		phds = append(phds, "?")
		sets = append(sets, "`"+col+"` = ?")
	}
	params = append(params, params...)
	query := "INSERT INTO `" + table + "` (`" + StrImplode("`, `", cols...) + "`) VALUES (" + StrImplode(", ", phds...) + ") "
	query += "ON DUPLICATE KEY UPDATE " + StrImplode(", ", sets...)
	db, err := m.GetDB(false)
	if err != nil {
		return 0, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), m.TimeoutInsert)
	defer cancel()
	defer func() {
		m.AddLog(query, params, err)
	}()
	if res, err := db.ExecContext(ctx, query, params...); err != nil {
		return 0, err
	} else {
		return res.LastInsertId()
	}
}

// InsertIgnore 插入，如果存在就忽略
func (m *Mysql) InsertIgnore(table string, row *MysqlRow) (int64, error) {
	return m.Insert(table, row, true)
}

// InsertReplace 插入，如果存在就替换
func (m *Mysql) InsertReplace(table string, row *MysqlRow) (int64, error) {
	return m.Insert(table, row, false, true)
}

// Insert 插入一条数据
func (m *Mysql) Insert(table string, row *MysqlRow, ignoreOrReplace ...bool) (int64, error) {
	ignore := false
	replace := false
	switch len(ignoreOrReplace) {
	case 0:
		ignore = false
		replace = false
	case 1:
		ignore = ignoreOrReplace[0]
		replace = false
	case 2:
		ignore = ignoreOrReplace[0]
		replace = ignoreOrReplace[1]
	default:
		ignore = ignoreOrReplace[0]
		replace = ignoreOrReplace[1]
	}
	if row == nil || *row == nil || len(*row) == 0 {
		return 0, fmt.Errorf("row is empty")
	}
	cols := make([]string, 0)
	phds := make([]string, 0)
	params := make([]interface{}, 0)
	for col, param := range *row {
		cols = append(cols, col)
		params = append(params, param)
		phds = append(phds, "?")
	}
	query := ""
	if ignore {
		query = "INSERT IGNORE "
	} else if replace {
		query = "REPLACE "
	} else {
		query = "INSERT "
	}
	query += "INTO `" + table + "` (`" + StrImplode("`, `", cols...) + "`) VALUES (" + StrImplode(", ", phds...) + ")"
	db, err := m.GetDB(false)
	if err != nil {
		return 0, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), m.TimeoutInsert)
	defer cancel()
	defer func() {
		m.AddLog(query, params, err)
	}()
	if res, err := db.ExecContext(ctx, query, params...); err != nil {
		return 0, err
	} else {
		return res.LastInsertId()
	}
}

// InsertBatch 批量插入
func (m *Mysql) InsertBatch(table string, rows []*MysqlRow) (ret int64, err error) {
	if len(rows) == 0 {
		return 0, fmt.Errorf("data is empty")
	}
	cols := make([]string, 0)
	phds := make([]string, 0)
	colsReal := make([]string, 0)
	for col := range *(rows[0]) {
		cols = append(cols, col)
		colsReal = append(colsReal, col)
		phds = append(phds, "?")
	}
	nc := len(cols)
	if nc == 0 {
		return 0, fmt.Errorf("wrong data format")
	}
	query := "INSERT IGNORE INTO `" + table + "` (`" + StrImplode("`, `", cols...) + "`) VALUES (" + StrImplode(", ", phds...) + ")"
	db, err := m.GetDB(false)
	if err != nil {
		return 0, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), m.TimeoutInsert)
	defer cancel()
	stmt, err := db.PrepareContext(ctx, query)
	if err != nil {
		return 0, err
	}
	defer stmt.Close()
	logs := make([]interface{}, 0)
	defer func() {
		m.Logger.Push(logs...)
	}()
	for _, row := range rows {
		params := make([]interface{}, 0)
		for _, col := range colsReal {
			params = append(params, (*row)[col])
		}
		if res, err := stmt.ExecContext(ctx, params...); err == nil {
			if affected, err := res.RowsAffected(); err == nil {
				ret += affected
			} else {
				return ret, err
			}
		} else {
			return ret, err
		}
		logs = append(logs, &QueryLog{
			When:   TimeNowNice(),
			Query:  query,
			Params: params,
			Err:    err,
		})
	}
	return ret, nil
}

// Select 查询数据
func (m *Mysql) Select(query string, params ...interface{}) (rows []*MysqlRow, err error) {
	err = m.SelWalk(func(row *MysqlRow) error {
		rows = append(rows, row)
		return nil
	}, query, params...)
	return
}

// SelPage 按页查询
func (m *Mysql) SelPage(page, numPerPage int64, table, queryWhere, queryOrder, selCols string, fn func(row *MysqlRow) error, params ...interface{}) (totalItem, totalPage, curPage int64, err error) {
	if StrTrim(queryWhere) == "" {
		queryWhere = "1 = 1"
	}
	qCount := "SELECT COUNT(*) FROM `" + table + "` WHERE " + queryWhere
	totalItem, err = m.Count(qCount, params...)
	if err != nil {
		return
	}
	if totalItem == 0 {
		totalPage = 0
		curPage = 1
		return
	}
	if numPerPage < 1 {
		numPerPage = 1
	}
	totalPage = int64(math.Ceil(float64(totalItem) / float64(numPerPage)))
	if page < 1 {
		page = 1
	}
	curPage = page
	if curPage > totalPage {
		return
	}
	if StrTrim(selCols) == "" {
		selCols = "*"
	}
	if StrTrim(queryOrder) != "" {
		queryOrder = " ORDER BY " + queryOrder
	} else {
		queryOrder = ""
	}
	qSelect := "SELECT " + selCols + " FROM `" + table + "` WHERE " + queryWhere + queryOrder + " LIMIT ?, ?"
	params = append(params, (curPage-1)*numPerPage, numPerPage)
	err = m.SelWalk(fn, qSelect, params...)
	return
}

// SelWalk 查询并逐条处理
func (m *Mysql) SelWalk(fn func(row *MysqlRow) error, query string, params ...interface{}) error {
	if strings.ToLower(query[0:7]) != "select " || len(query) < 8 {
		return fmt.Errorf("wrong select sql")
	}
	db, err := m.GetDB(true)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), m.TimeoutSelect)
	defer cancel()
	defer func() {
		m.AddLog(query, params, err)
	}()
	rows, err := db.QueryContext(ctx, query, params...)
	if err != nil {
		return err
	}
	defer rows.Close()
	cols, err := rows.Columns()
	if err != nil {
		return err
	}
	colsLen := len(cols)
	for rows.Next() {
		rawBuffers := make([]sql.RawBytes, colsLen)
		scanArgs := make([]interface{}, colsLen)
		for i := range rawBuffers {
			scanArgs[i] = &rawBuffers[i]
		}
		if err = rows.Scan(scanArgs...); err != nil {
			return err
		}
		row := &MysqlRow{}
		for i, bs := range rawBuffers {
			(*row)[cols[i]] = string(bs)
		}
		if err = fn(row); err != nil {
			return err
		}
	}
	return nil
}

// GetOne 按ID取一条数据
func (m *Mysql) GetOne(table, pk, id string, cols ...string) (*MysqlRow, error) {
	qCols := "*"
	if len(cols) > 0 {
		qCols = "`" + StrImplode("`, `", cols...) + "`"
	}
	if StrTrim(pk) == "" {
		pk = "id"
	}
	rows, err := m.Select("SELECT "+qCols+" FROM `"+table+"` WHERE `"+pk+"` = ? LIMIT 1", id)
	if err != nil {
		return nil, err
	}
	if len(rows) > 0 {
		return rows[0], nil
	}
	return nil, nil
}

// IdsWalk 按多个ID取数据并逐条处理
func (m *Mysql) IdsWalk(fn func(row *MysqlRow) error, table string, pk string, ids []interface{}, cols ...string) error {
	idsNum := len(ids)
	if idsNum == 0 {
		return fmt.Errorf("no id")
	}
	queryCols := "*"
	if len(cols) > 0 {
		queryCols = "`" + StrImplode("`, `", cols...) + "`"
	}
	phds := strings.TrimRight(strings.Repeat(" ?,", idsNum), ",")
	if StrTrim(pk) == "" {
		pk = "id"
	}
	return m.SelWalk(func(row *MysqlRow) error {
		return fn(row)
	}, "SELECT "+queryCols+" FROM `"+table+"` WHERE `"+pk+"` IN ("+phds+" )", ids...)
}

// GetByIds 按多个ID取数据
func (m *Mysql) GetByIds(table, pk string, ids []interface{}, cols ...string) (rows []*MysqlRow, err error) {
	switch len(ids) {
	case 0:
		err = fmt.Errorf("no id")
		return
	case 1:
		if row, err := m.GetOne(table, pk, InterfaceToStr(ids[0]), cols...); err != nil {
			return rows, err
		} else if row != nil {
			rows = append(rows, row)
			return rows, nil
		}
		return
	default:
		err = m.IdsWalk(func(row *MysqlRow) error {
			rows = append(rows, row)
			return nil
		}, table, pk, ids, cols...)
		return
	}
}

// Count 查询数据结果个数
func (m *Mysql) Count(query string, params ...interface{}) (cnt int64, err error) {
	if strings.ToLower(query[0:13]) != "select count(" || len(query) < 14 {
		return 0, fmt.Errorf("wrong select count sql")
	}
	db, err := m.GetDB(true)
	if err != nil {
		return 0, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), m.TimeoutSelect)
	defer cancel()
	defer func() {
		m.AddLog(query, params, err)
	}()
	err = db.QueryRowContext(ctx, query, params...).Scan(&cnt)
	return
}

// Update 更新数据
func (m *Mysql) Update(query string, params ...interface{}) (int64, error) {
	if sqlStart7 := strings.ToLower(query[0:7]); (sqlStart7 != "update " && sqlStart7 != "delete ") || len(query) < 8 {
		return 0, fmt.Errorf("wrong update or delete sql")
	}
	db, err := m.GetDB(false)
	if err != nil {
		return 0, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), m.TimeoutUpdate)
	defer cancel()
	defer func() {
		m.AddLog(query, params, err)
	}()
	if res, err := db.ExecContext(ctx, query, params...); err != nil {
		return 0, err
	} else if affected, err := res.RowsAffected(); err != nil {
		return 0, err
	} else {
		return affected, nil
	}
}

// UpdateById 按ID更新数据
func (m *Mysql) UpdateById(table string, row *MysqlRow, pk, id string) (int64, error) {
	if row == nil || *row == nil {
		return 0, fmt.Errorf("empty row")
	}
	colsNum := len(*row)
	id = StrTrim(id)
	if colsNum == 0 || id == "" {
		return 0, fmt.Errorf("empty row or id")
	}
	sets := make([]string, 0)
	params := make([]interface{}, 0)
	if StrTrim(pk) == "" {
		pk = "id"
	}
	for col, param := range *row {
		if col == pk {
			continue
		}
		sets = append(sets, "`"+col+"` = ?")
		params = append(params, param)
	}
	if len(sets) == 0 {
		return 0, fmt.Errorf("wrong row")
	}
	params = append(params, id)
	return m.Update("UPDATE `"+table+"` SET "+StrImplode(", ", sets...)+" WHERE `"+pk+"` = ?", params...)
}

// Delete 删除数据
func (m *Mysql) Delete(query string, params ...interface{}) (int64, error) {
	if strings.ToLower(query[0:7]) != "delete " || len(query) < 8 {
		return 0, fmt.Errorf("wrong delete sql")
	}
	return m.Update(query, params...)
}

// DelByIds 按一个或多个ID删除数据
func (m *Mysql) DelByIds(table, pk string, ids ...interface{}) (i int64, err error) {
	idsNum := len(ids)
	if idsNum == 0 {
		return 0, fmt.Errorf("no id")
	}
	if StrTrim(pk) == "" {
		pk = "id"
	}
	if idsNum == 1 {
		if ids[0] == "" {
			return 0, fmt.Errorf("id is empty")
		}
		i, err = m.Delete("DELETE FROM `"+table+"` WHERE `"+pk+"` = ?", ids[0])
	} else {
		phds := strings.TrimRight(strings.Repeat(" ?,", idsNum), ",")
		i, err = m.Delete("DELETE FROM `"+table+"` WHERE `"+pk+"` IN ("+phds+" )", ids...)
	}
	return
}

// Transaction 事务
func (m *Mysql) Transaction(fn func(tx *sql.Tx, addLog func(query string, params []interface{}, err error)) error, opts *sql.TxOptions) error {
	db, err := m.GetDB(false)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), m.TimeoutTrans)
	defer cancel()
	tx, err := db.BeginTx(ctx, opts)
	if err != nil {
		return err
	}
	defer func() {
		if p := recover(); p != nil {
			_ = tx.Rollback()
			panic(p)
		}
		if err != nil {
			_ = tx.Rollback()
		}
	}()
	if err = fn(tx, m.AddLog); err == nil {
		err = tx.Commit()
	}
	return err
}

// ToStr 查询结果转字符串
func (row *MysqlRow) ToStr(key string) string {
	return InterfaceToStr((*row)[key])
}

// ToInt64 查询结果转整形
func (row *MysqlRow) ToInt64(key string) int64 {
	return InterfaceToInt64((*row)[key])
}

// ToNum 查询结果转数字
func (row *MysqlRow) ToNum(key string) float64 {
	return InterfaceToFloat64((*row)[key])
}

// AddLog 添加日志
func (m *Mysql) AddLog(query string, params []interface{}, err error) {
	if m.Listener != nil {
		m.Listener(err, query, params...)
		return
	}
	m.Logger.Push(&MysqlLog{
		When:   TimeNowNice(),
		Query:  query,
		Params: params,
		Err:    err,
	})
}

// PopLogs 处理日志
func (m *Mysql) PopLogs(fn func(ql *MysqlLog) error) error {
	for {
		if log, _ := m.Logger.Pop(); log == nil {
			break
		} else if ql, ok := log.(*MysqlLog); ok {
			if err := fn(ql); err != nil {
				return err
			}
		}
	}
	return nil
}
